package com.it.state;

import com.it.operator.utils.SourceUtils;
import com.it.pojo.Event;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * 使用MapState来模拟时间窗口的功能.
 *
 * @author code1997
 */
public class FakeWindowDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        SingleOutputStreamOperator<Event> eventSource = SourceUtils.getEventSource(executionEnvironment);
        eventSource.print("input");
        eventSource.keyBy(data -> data.url)
                .process(new FakeWindowFunction(5000L))
                .print();
        executionEnvironment.execute();
    }

    public static class FakeWindowFunction extends KeyedProcessFunction<String, Event, String> {

        Long windowSize;
        //保存窗口的统计数据
        MapState<Long, Long> fakeWindows;

        public FakeWindowFunction(Long ts) {
            windowSize = ts;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            fakeWindows = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Long>("fake_windows", Types.LONG, Types.LONG));
        }

        @Override

        public void processElement(Event value, KeyedProcessFunction<String, Event, String>.Context ctx, Collector<String> out) throws Exception {
            //计算窗口的范围
            Long windowStart = value.timestamp / windowSize * windowSize;
            long windowEnd = windowStart + windowSize;
            //注册定时器
            ctx.timerService().registerEventTimeTimer(windowEnd - 1);
            //更新状态，进行增量聚合
            Long oldCount = fakeWindows.get(windowStart);
            fakeWindows.put(windowStart, oldCount == null ? 1 : oldCount + 1);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            long windowEnd = timestamp + 1;
            long windStart = windowEnd - windowSize;
            String currentUrl = ctx.getCurrentKey();
            Long count = fakeWindows.get(windStart);
            //时间到了，进行结果的收集
            out.collect("窗口:" + new Timestamp(windStart) + "~" + new Timestamp(windowEnd) + ", url:" + currentUrl + " 的点击数为:" + count);
            //模拟窗口关闭，删除对应的状态
            fakeWindows.remove(windStart);
        }

    }
}
